package registry

import (
	"encoding/json"
	"errors"
	"fmt"
	"github.com/go-zookeeper/zk"
	"maps"
	"path"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"
	"wf-deploy-agent/conf"
	"wf-deploy-agent/infra"
	"wf-deploy-agent/registry/entity"
	"wf-deploy-agent/util"
	"wf-deploy-agent/zookeeper"
)

var ms, mw sync.Mutex

// ZkDiscovery 实现了 Discovery
type ZkDiscovery struct {
	Client   *zookeeper.Client
	Services map[string][]*registry.ServiceInstance
	Watchers map[string]any
}

func NewZkDiscovery() *ZkDiscovery {
	d := &ZkDiscovery{}
	d.Watchers = map[string]any{}
	d.Services = map[string][]*registry.ServiceInstance{}
	d.Client = zookeeper.New()
	return d
}

func (d *ZkDiscovery) GetAbsPath(node string) string {
	if node == "" {
		infra.Logger.Errorln("node path is empty")
	}

	fullPath := path.Join("/", conf.CF.Registry.Root, node)
	if strings.HasSuffix(fullPath, "/") {
		fullPath = fullPath[0 : len(fullPath)-1]
	}

	return fullPath
}

// Fetch 拉群服务实例
func (d *ZkDiscovery) Fetch() {
	items, err := d.GetCatalog()
	if err != nil {
		return
	}

	for _, v := range items {
		serviceNode := path.Join("/", conf.CF.Registry.Root, v)
		d.SyncInstance(serviceNode)
	}
}

// GetCatalog 获取服务目录
func (d *ZkDiscovery) GetCatalog() ([]string, error) {
	items, err := d.Client.GetChildren(path.Join("/" + conf.CF.Registry.Root))
	return items, err
}

// GetService 获取服务实例
func (d *ZkDiscovery) GetService(name string) ([]*registry.ServiceInstance, error) {
	instances := d.Services[name]
	if instances != nil && len(instances) > 0 {
		return instances, nil
	}

	serviceNode := d.GetAbsPath(name)
	d.SyncInstance(serviceNode)

	return d.Services[name], nil
}

// GetInstance 获取单一服务实例
func (d *ZkDiscovery) GetInstance(name string) (*registry.ServiceInstance, error) {
	instances, err := d.GetService(name)
	if err != nil {
		return nil, err
	}

	if len(instances) == 0 {
		return nil, errors.New(fmt.Sprintf("there is no avaiable instance for service[%s]", name))
	}

	now := time.Now().Unix()
	i := int(now) % len(instances)
	instance := instances[i]

	return instance, nil
}

// GetOrientedInstance 获取定向的单一服务实例
func (d *ZkDiscovery) GetOrientedInstance(name, routingKey string) (*registry.ServiceInstance, error) {
	instances, err := d.GetService(name)
	if err != nil {
		return nil, err
	}

	if len(instances) == 0 {
		return nil, errors.New(fmt.Sprintf("there is no avaiable instance for service[%s]", name))
	}

	hash := util.GetHashCode(routingKey)
	i := hash % len(instances)
	instance := instances[i]

	return instance, nil
}

// Monitor 监控服务实例
func (d *ZkDiscovery) Monitor() {
	d.Fetch()
	go d.watchRootNode()
	go d.watchExistedService()
}

func (d *ZkDiscovery) watchRootNode() {
	for {
		_, _, events, err := d.Client.GetConn().ChildrenW(conf.CF.Registry.Root)
		if err != nil {
			infra.Logger.Errorln(err)
			continue
		}

		for event := range events {
			switch event.Type {
			case zk.EventNodeChildrenChanged:
				//go d.Fetch()
				d.watchExistedService()
			default:
			}
		}

		<-events
	}
}

func (d *ZkDiscovery) watchExistedService() {
	catalogs, err := d.GetCatalog()
	if err != nil {
		infra.Logger.Errorln(err)
		return
	}

	//清除已经不存在的服务
	d.clearService(catalogs)
	//清除已经不存在的服务监控标记
	d.clearWatcher(catalogs)

	for _, v := range catalogs {
		if d.existsWatcher(v) {
			continue
		}

		serviceNode := path.Join("/", conf.CF.Registry.Root, v)
		go d.watchServiceNode(serviceNode)
	}
}

// 清除已经不存在的服务
func (d *ZkDiscovery) clearService(catalogs []string) {
	ms.Lock()
	maps.DeleteFunc(d.Services, func(key string, instances []*registry.ServiceInstance) bool {
		if !slices.Contains(catalogs, key) {
			return true
		}
		return false
	})
	ms.Unlock()
}

// 清除已经不存在的服务监控标记
func (d *ZkDiscovery) clearWatcher(catalogs []string) {
	mw.Lock()
	maps.DeleteFunc(d.Watchers, func(key string, value any) bool {
		if !slices.Contains(catalogs, key) {
			return true
		}
		return false
	})
	mw.Unlock()
}

// 验证实例端口是否通常
func (d *ZkDiscovery) validate(instance *registry.ServiceInstance) bool {
	return util.IsOpened(instance.Host, fmt.Sprint(instance.Port))
}

// 删除服务实例
func (d *ZkDiscovery) removeService(serviceName string) {
	ms.Lock()
	delete(d.Services, serviceName)
	ms.Unlock()
}

// 更新服务实例
func (d *ZkDiscovery) updateService(serviceName string, instances []*registry.ServiceInstance) {
	ms.Lock()
	d.Services[serviceName] = instances
	ms.Unlock()
}

// 是否存在服务监控标记
func (d *ZkDiscovery) existsWatcher(serviceName string) bool {
	_, ok := d.Watchers[serviceName]
	return ok
}

// 删除服务监控标记
func (d *ZkDiscovery) removeWatcher(serviceName string) {
	mw.Lock()
	delete(d.Watchers, serviceName)
	mw.Unlock()
}

// 更新服务监控标记
func (d *ZkDiscovery) updateWatcher(serviceName string) {
	mw.Lock()
	d.Watchers[serviceName] = 1
	mw.Unlock()
}

func (d *ZkDiscovery) SyncInstance(serviceNode string) {
	exists, err := d.Client.Exists(serviceNode)
	if err != nil || !exists {
		return
	}

	instanceNodes, err := d.Client.GetChildren(serviceNode)
	if err != nil {
		if err != zk.ErrNoNode {
			infra.Logger.Errorln(err)
		}

		return
	}

	instances := make([]*registry.ServiceInstance, 0)
	for _, v := range instanceNodes {
		instanceNode := path.Join(serviceNode, v)
		bytes, err := d.Client.GetValue(instanceNode)
		if err != nil {
			if err != zk.ErrNoNode {
				infra.Logger.Errorln(err)
			}
			continue
		}

		registration := &registry.ServiceRegistration{}
		err = json.Unmarshal(bytes, registration)
		if err != nil {
			infra.Logger.Errorln(err)
			continue
		}

		instance := &registry.ServiceInstance{
			InstanceId: registration.Id,
			ServiceId:  registration.Name,
			Host:       registration.Address,
			Port:       registration.Port,
		}

		if registration.SslPort != nil {
			instance.Secure = true
		}

		if d.validate(instance) {
			instances = append(instances, instance)
		}
	}

	//节点下存在服务实例就更新缓存，否则就删除对应的缓存
	serviceName := strings.ReplaceAll(serviceNode, conf.CF.Registry.Root+"/", "")
	if len(instances) == 0 {
		d.removeService(serviceName)
		d.removeWatcher(serviceName)
	} else {
		sort.Slice(instances, func(i, j int) bool {
			return instances[i].GetHashCode() < instances[j].GetHashCode()
		})

		d.updateService(serviceName, instances)
	}
}

func (d *ZkDiscovery) watchServiceNode(serviceNode string) {
	for {
		exists, err := d.Client.Exists(serviceNode)
		if err != nil {
			continue
		}

		if !exists {
			return
		}

		_, _, events, err := d.Client.GetConn().ChildrenW(serviceNode)
		if err != nil {
			if err != zk.ErrNoNode {
				infra.Logger.Errorln(err)
			}
			continue
		}

		serviceName := strings.ReplaceAll(serviceNode, conf.CF.Registry.Root+"/", "")
		d.updateWatcher(serviceName)

		for event := range events {
			switch event.Type {
			case zk.EventNodeCreated:
				infra.Logger.Debug("Node created.")
				// Do somthing
			case zk.EventNodeDeleted:
				infra.Logger.Debug("Node deleted!")
				// Do somthing
			case zk.EventNodeDataChanged:
				infra.Logger.Debug("Node changed.")
				// Do somthing
			case zk.EventNodeChildrenChanged:
				infra.Logger.Debug("Node children changed.")
				d.SyncInstance(serviceNode)
				// Do somthing
			default:
				infra.Logger.Debug("Something happen.")
			}
		}

		<-events
	}
}
